package com.binarysoft.service.common.ping;

import java.io.IOException;
import java.io.InterruptedIOException;
import java.net.InetAddress;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

import org.savarese.vserv.tcpip.ICMPEchoPacket;
import org.savarese.vserv.tcpip.ICMPPacket;
import org.savarese.vserv.tcpip.OctetConverter;

import com.savarese.rocksaw.net.RawSocket;
import common.framework.dsb.AbstractServiceBean;
import common.framework.dsb.ServiceException;
import common.framework.dsb.annotation.DSBService;
import common.framework.dsb.service.ServiceContext;
import common.framework.log.Logger;

/**
 * （1）一共两个线程：发送线程和接收线程，两个队列：等待队列和执行队列，和一个socket。<br/>
 * <br/>
 * （2）先把ping任务加到等待队列并让调用线程等待ping任务对象的锁。<br/>
 * 发送线程负责：按照FIFO的顺序把ping任务从等待队列取出，将ping任务加到执行队列中，然后发送ICMP包。
 * 接收线程负责：按照接收到的ICMP回复从执行队列中找出ping任务，保存结果，然后唤醒等待中的调用线程（返回结果）。<br/>
 * <br/>
 * （3）ping结果的rtt计算：开始时间在echo回复包中获取，结束时间为接收到回复包时系统的当前时间。rtt = 结束时间 - 开始时间<br/>
 * ttl从echo回复包中获取<br/>
 * <br/>
 * 
 * @author Li Xu
 * 
 */
@DSBService(name = "DSB/PingService", startup = true)
public class PingServiceBean extends AbstractServiceBean implements PingService {

	private static final int DefaultTimeout = 3000;
	private static final int DefualtDataByteLength = 32;
	private static final int IpHeaderByteLength = 5;
	private static final int PacketHeaderByteLength = 28;

	/** 服务是否关闭的标志 */
	private boolean isServiceClosed = false;

	/** 线程退出同步模块,只有发送,接收线程都退出,ping服务才会退出 */
	private CountDownLatch waitChildThreadClose = new CountDownLatch(2);

	/** 等待队列 */
	// private ArrayBlockingQueue<PingTask> pendingTaskQueue = new
	// ArrayBlockingQueue<PingTask>(5000);
	private LinkedList<PingTask> pendingTaskQueue = new LinkedList<PingTask>();

	/** 执行队列 */
	private Map<Long, PingTask> processingTaskQueue = new HashMap<Long, PingTask>();

	/** 系统中仅一个socket对象 */
	private RawSocket socket = null;

	/** 用于分配每个PING标识号,每个ping有一个唯一的Ping标识号 */
	private final static AtomicInteger pingIdentifier = new AtomicInteger(0);

	/** 用于分配每个设备发送的ping任务的序列号,每个任务后序号自增，由于ping协议编码的限制,ping序列号到达65535后,循环 */
	private final static AtomicInteger pingSequence = new AtomicInteger(0);

	/**
	 * init
	 */
	@Override
	public void start(ServiceContext serviceContext) throws Exception {
		super.start(serviceContext);
		try {
			serviceContext.loadConfigProperties("ping.properties");
			this.isServiceClosed = false;
			this.socket = new RawSocket();

			this.socket.open(RawSocket.PF_INET, RawSocket.getProtocolByName("icmp"));
			this.socket.bind(InetAddress.getByName(serviceContext.getProperty("host")));

			this.socket.setSendTimeout(DefaultTimeout);
			this.socket.setReceiveTimeout(DefaultTimeout);

			System.out.println("ping.src.host=" + serviceContext.getProperty("host"));
		} catch (Throwable th) {
			th.printStackTrace(System.out);
		}

		new ReceiveThread(this).start();
		new SendThread(this).start();

	}

	@Override
	public void refresh() throws Exception {
		// TODO Auto-generated method stub

	}

	@Override
	public PingResult doPing(InetAddress target, int timeout, int dataByteLength) {
		PingResult pingResult = null;
		int id = this.getPingIdentifier();
		int seq = this.getPingSequence();
		ICMPEchoPacket sendPacket;
		PingTask pingTask = null;
		int dataOffset, offset, packetByteLength;

		/** receivePacket */
		sendPacket = new ICMPEchoPacket(1);// icmp send packet
		byte[] sendData = new byte[dataByteLength + PacketHeaderByteLength];
		byte[] recvData = new byte[dataByteLength + PacketHeaderByteLength];
		sendPacket.setData(sendData);
		sendPacket.setIPHeaderLength(IpHeaderByteLength);
		sendPacket.setICMPDataByteLength(dataByteLength);
		offset = sendPacket.getIPHeaderByteLength();
		dataOffset = offset + sendPacket.getICMPHeaderByteLength();// dataOffset
		// = 8 + 20
		packetByteLength = sendPacket.getICMPPacketByteLength();
		sendPacket.setCode(0);
		sendPacket.setIdentifier(id);
		sendPacket.setSequenceNumber(seq);
		sendPacket.setType(ICMPPacket.TYPE_ECHO_REQUEST);

		/** pingTask */
		pingTask = new PingTask();
		long taskKey = id * 100000 + seq;
		pingTask.setDataOffset(dataOffset);
		pingTask.setDataByteLength(dataByteLength);
		pingTask.setOffset(offset);
		pingTask.setPacketByteLength(packetByteLength);
		pingTask.setRecvData(recvData);
		pingTask.setSendData(sendData);
		pingTask.setSendPacket(sendPacket);
		pingTask.setTarget(target);
		pingTask.setTimeout(timeout);//
		pingTask.setTaskKey(taskKey);

		/** add pingTask into pendingTaskQueue */
		addPendingPingTask(pingTask);

		/** wait for reply */
		synchronized (pingTask) {
			try {
				pingTask.wait(timeout);// 超时控制
			} catch (InterruptedException e) {
				removeProcessingPingTask(taskKey);
				// e.printStackTrace();
			}
		}

		/** has got the reply or timeout */
		pingResult = new PingResult();
		if (pingTask.getHaveReply()) {// has got the reply
			pingTask.getRecvPacket().getData(recvData);// copy
			// 回复包的内容到recvData[]，用来取发包的开始时间
			pingTask.setStartTime(OctetConverter.octetsToLong(recvData, dataOffset));
			double rtt = (pingTask.getEndTime() - pingTask.getStartTime()) / 1e6; // 计算接收延时时间
			int ttl = pingTask.getRecvPacket().getTTL(); // 取出ttl信息
			if (ttl < 0)
				ttl = 256 + ttl;
			pingResult.setTimedout(false);
			pingResult.setRtt(rtt);
			pingResult.setTtl(ttl);
		} else {
			/**
			 * delete pingTask from processingTaskQueue
			 * 两种情况：如果接到回复，则在接收线程中删除任务(receiveEchoReply方法中实现
			 * )。如果没有接到回复(超时)，则在此处删除
			 */
			removeProcessingPingTask(taskKey);
			pingResult.setTimedout(true);
		}
		pingResult.setDataByteLength(dataByteLength);
		pingResult.setTarget(target);

		return pingResult;
	}

	@Override
	public PingResult doPing(InetAddress target, int timeout) {
		return doPing(target, timeout, DefualtDataByteLength);
	}

	@Override
	public PingResult doPing(InetAddress target) {
		return doPing(target, DefaultTimeout, DefualtDataByteLength);
	}

	public void sendEchoRequest(PingTask pingTask) {
		try {
			OctetConverter.longToOctets(System.nanoTime(), pingTask.getSendData(), pingTask.getDataOffset());
			pingTask.getSendPacket().computeICMPChecksum();// 必须先计算校验和
			socket.write(pingTask.getTarget(), pingTask.getSendData(), pingTask.getOffset(), pingTask.getPacketByteLength());
			// System.out.println("[PingServiceBean] -" +
			// pingTask.getSendPacket().getIdentifier() + "- have sent a ICMP
			// packet");
		} catch (IllegalArgumentException e) {// 异常处理
			synchronized (pingTask) {
				pingTask.setHaveReply(false);
				pingTask.notifyAll();
			}
			Logger.log(Logger.LOW_LEVEL, e.getMessage());
			// e.printStackTrace(System.out);
		} catch (InterruptedIOException e) {
			synchronized (pingTask) {
				pingTask.setHaveReply(false);
				pingTask.notifyAll();
			}
			Logger.log(Logger.LOW_LEVEL, e.getMessage());
		} catch (IOException e) {
			synchronized (pingTask) {
				pingTask.setHaveReply(false);
				pingTask.notifyAll();
			}
			Logger.log(Logger.LOW_LEVEL, e.getMessage());
		}
	}

	/**
	 * 接收ping响应信息的处理逻辑: 1 接收ehco回复包 2 取系统当前时间（用于计算returnTime） 3 从echo回复包中读出ping
	 * id，seq，并从processingTaskQueue中取出task 4 把echo回复包保存到task（用于取结果）
	 * 
	 */
	public void receiveEchoReply() {
		PingTask pingTask = null;
		try {
			// System.out.println("[PingServiceBean] call receiveEchoReply()");
			// packet
			ICMPEchoPacket recvPacket = new ICMPEchoPacket(1); // icmp recv
			recvPacket.setType(ICMPPacket.TYPE_ECHO_REPLY);
			byte[] recvData = new byte[256];// 用于暂存echo回复，因为先要读出包才能读出id
			recvPacket.setData(recvData);
			recvPacket.setIPHeaderLength(IpHeaderByteLength);
			long endTime = 0;
			do {// 每接收到一个包循环只执行一次
				this.socket.read(recvData); // 阻塞式的接收
				endTime = System.nanoTime();// 取系统当前时间，用于计算returnTime
			} while (recvPacket.getType() != ICMPPacket.TYPE_ECHO_REPLY);
			int identifier = recvPacket.getIdentifier(); // 取出ping包的标识,返回信息中有一个相同的标识
			int sequence = recvPacket.getSequenceNumber();
			pingTask = removeProcessingPingTask(identifier * 100000 + sequence);// 根据key从processingTaskQueue中取出task
			if (pingTask != null) {
				pingTask.setRecvPacket(recvPacket);
				pingTask.setEndTime(endTime);
				pingTask.setHaveReply(true);
				synchronized (pingTask) {
					pingTask.notifyAll();// 唤醒用户调用线程，以返回结果
				}
			}
		} catch (InterruptedIOException e) {// 异常处理
			if (pingTask != null) {
				synchronized (pingTask) {
					pingTask.setHaveReply(false);
					pingTask.notifyAll();
				}
			}
			Logger.log(Logger.LOW_LEVEL, e.getMessage());
			// e.printStackTrace(System.out);
		} catch (IOException e) {
			if (pingTask != null) {
				synchronized (pingTask) {
					pingTask.setHaveReply(false);
					pingTask.notifyAll();
				}
			}
			Logger.log(Logger.LOW_LEVEL, e.getMessage());
			// e.printStackTrace(System.out);
		}

	}

	/**
	 * 供服务调用者调用。添加到队列尾部
	 * 
	 * @return
	 */
	private void addPendingPingTask(PingTask pingTask) {
		synchronized (pendingTaskQueue) {
			pendingTaskQueue.addLast(pingTask);
			// System.out.println("[PingServiceBean] ************* after
			// addPendingPingTask. PendingPingTask size: " +
			// pendingTaskQueue.size());
			pendingTaskQueue.notifyAll();
		}
	}

	/**
	 * 供发送线程调用. 调用removeFirst()，即取队列头并删除。
	 * 
	 * @return
	 */
	public PingTask getPendingPingTask() {
		PingTask task = null;

		synchronized (pendingTaskQueue) {
			if (pendingTaskQueue.size() > 0) {
				task = pendingTaskQueue.removeFirst();
				// System.out.println("[PingServiceBean] ************* after
				// getPendingPingTask. PendingPingTask size: " +
				// pendingTaskQueue.size());
			} else {
				try {
					pendingTaskQueue.wait();// 如果队列为空，扫描线程等待，直到队列不为空才唤醒
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
		return task;
	}

	/**
	 * 供发送线程调用。 把pingTask从pendingTaskQueue中移到processingTaskQueue
	 * 
	 * @return
	 */
	public void addProcessingPingTask(PingTask pingTask) {
		synchronized (processingTaskQueue) {
			processingTaskQueue.put(pingTask.getTaskKey(), pingTask);
			// System.out.println("[PingServiceBean] ************* after
			// addProcessingPingTask. ProcessingPingTask size: " +
			// processingTaskQueue.size());
			processingTaskQueue.notifyAll();
		}
	}

	/**
	 * 供接收线程和服务调用者调用。 即，待取出结果后从processingTaskQueue中删除
	 * 
	 * @return
	 */
	public PingTask removeProcessingPingTask(long key) {
		PingTask pingTask = null;
		synchronized (processingTaskQueue) {
			if (processingTaskQueue.size() > 0) {
				pingTask = processingTaskQueue.remove(key);
			}
		}
		return pingTask;
	}

	@Override
	public void close() throws ServiceException {
		this.isServiceClosed = true;
		try {
			this.socket.close(); // 关闭rawSocket
		} catch (IOException e) {
			Logger.log(Logger.FUNCTION_LEVEL, "[PingServiceBean] RawSocket关闭错误");
		}

		Logger.log(Logger.FUNCTION_LEVEL, "[PingServiceBean] sevice closed");
		//
		// System.out.println("[PingServiceBean] sevice closed");
	}

	public boolean isServiceClosed() {
		return this.isServiceClosed;
	}

	public boolean setServiceClosed(boolean isServiceClosed) {
		return this.isServiceClosed = isServiceClosed;
	}

	/** 线程同步控制器的动作 */
	public void countDown() {
		this.waitChildThreadClose.countDown();// 减1
	}

	/**
	 * 返回ping的标识
	 * 
	 * @return ping的标识,范围1-65535
	 */
	public int getPingIdentifier() {
		int id = pingIdentifier.getAndIncrement();
		id = id % 65535;
		return id;
	}

	/**
	 * 返回ping的序列号
	 * 
	 * @return ping的序列号,范围1-65535
	 */
	public int getPingSequence() {
		int seq = pingSequence.getAndIncrement();
		seq = seq % 65535;
		return seq;
	}

}
